package com.ruoyi.iot.service.impl;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import com.alibaba.fastjson.JSON;
import com.ruoyi.common.exception.ServiceException;
import com.ruoyi.iot.domain.Channel;
import com.ruoyi.iot.domain.emun.DevicePropertyDataRequest;
import org.apache.iotdb.isession.pool.SessionDataSetWrapper;
import org.apache.iotdb.rpc.IoTDBConnectionException;
import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.session.pool.SessionPool;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.read.common.Field;
import org.apache.tsfile.read.common.RowRecord;
import org.springframework.stereotype.Service;

import com.ruoyi.iot.domain.DeviceData;
import com.ruoyi.iot.mapper.DeviceMapper;
import com.ruoyi.iot.service.IDeviceDataService;
import com.ruoyi.iotcommon.domain.DeviceDataMessage;
import com.ruoyi.iotcommon.utils.IotdbSqlUtils;

import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;

/**
 * 默认从iotdb中插入和获取设备数据
 * 想使用其他时序数据库的，可以实现该接口重新插入和查询
 */
@Slf4j
@Service
public class DeviceDataServiceImpl implements IDeviceDataService{

    // 缓存所有产品的通道和数据项属性，每个5分钟数据库更新一次,key = productKey
    public static final Map<String,List<Channel>> PRODUCT_CHANNEL_PROPERTY_CACHE = new ConcurrentHashMap<>();

    @Resource
    SessionPool sessionPool;

    @Resource
    DeviceMapper deviceMapper;

    @Override
    public Map<String, Object> lastDeviceData(DeviceData deviceData) {
        String tableName = IotdbSqlUtils.buildTableName(deviceData.getProductKey(), deviceData.getChannelKey(), deviceData.getDeviceKey());
        String sql = IotdbSqlUtils.selectLast(tableName);
        Map<String, Object> result = new HashMap<>();
        try (SessionDataSetWrapper wrapper = sessionPool.executeQueryStatement(sql)){
            while (wrapper.hasNext()) {
                RowRecord record = wrapper.next();
                List<Field> fields = record.getFields();
                List<String> columnNames = wrapper.getColumnNames();
                for (int i = 0; i < columnNames.size(); i++) {
                    if ("Time".equalsIgnoreCase(columnNames.get(i))) continue;
                    result.put(columnNames.get(i), fields.get(i).getObjectValue(fields.get(i).getDataType()));
                }
              }
           } catch (Exception e) {
                log.error(e.getMessage());
           }
           
           log.info("testLast-设备数据：{}",JSON.toJSONString(result));
           return result;
    }

    @Override
    public DevicePropertyDataRequest queryDevicePropertyData(DevicePropertyDataRequest request) {
        return QueryDevicePropertiesData(request, Collections.singletonList(request.getPropertyKey()));
    }

    @Override
    public DevicePropertyDataRequest QueryDevicePropertiesData(DevicePropertyDataRequest request, List<String> propertyKeys) {
        if (request.getStartTime() > request.getEndTime()) {
            throw new ServiceException("开始时间不能大于结束时间");
        }
        String tableName = IotdbSqlUtils.buildTableName(request.getProductKey(), 
        request.getChannelKey(), request.getDeviceKey());
            
            // 构建查询SQL
            String sql;
            if (propertyKeys == null || propertyKeys.isEmpty()) {
                // 查询所有属性
                sql = buildQueryAllPropertiesSql(tableName, request);
            } else {
                // 查询指定属性
                String properties = String.join(",", propertyKeys);
                sql = buildQuerySpecificPropertiesSql(tableName, properties, request);
            }
            
            try (SessionDataSetWrapper wrapper = sessionPool.executeQueryStatement(sql)) {
                List<Object> dataList = new ArrayList<>();
                long lastTimestamp = 0;
                
                while (wrapper.hasNext()) {
                    RowRecord record = wrapper.next();
                    Map<String, Object> dataMap = new HashMap<>();
                    dataMap.put("time", record.getTimestamp());
                    
                    List<Field> fields = record.getFields();
                    List<String> columnNames = wrapper.getColumnNames();
                    
                    for (int i = 0; i < columnNames.size(); i++) {
                        if ("Time".equalsIgnoreCase(columnNames.get(i))) continue;
                        dataMap.put(columnNames.get(i), fields.get(i).getObjectValue(fields.get(i).getDataType()));
                    }
                    
                    dataList.add(dataMap);
                    lastTimestamp = record.getTimestamp();
                }
                
                // 设置返回结果
                request.setData(dataList);
                
                // 判断是否有下一页
                if (!dataList.isEmpty() && dataList.size() >= request.getPageSize()) {
                    request.setNextValid(true);
                    request.setNextTime(lastTimestamp);
                } else {
                    request.setNextValid(false);
                    request.setNextTime(null);
                }
                
            } catch (Exception e) {
                log.error("查询设备属性数据失败: {}", e.getMessage(), e);
                request.setNextValid(false);
                request.setNextTime(null);
                request.setData(Collections.emptyList());
            }
            
            return request;
    }

    @Override
    public int queryDevicePropertyDataCount(DevicePropertyDataRequest request) {
        String tableName = IotdbSqlUtils.buildTableName(
                request.getProductKey(),
                request.getChannelKey(),
                request.getDeviceKey()
        );

        // 构建统计SQL
        String sql = IotdbSqlUtils.selectCount(
                tableName,
                request.getStartTime(),
                request.getEndTime()
        );

        try (SessionDataSetWrapper wrapper = sessionPool.executeQueryStatement(sql)) {
            if (wrapper.hasNext()) {
                RowRecord record = wrapper.next();
                // count(*)结果在第一个字段
                return record.getFields().get(0).getIntV();
            }
            return 0;
        } catch (IoTDBConnectionException e) {
            log.error("IoTDB连接异常: {}", e.getMessage(), e);
            return 0;
        } catch (StatementExecutionException e) {
            log.error("查询执行失败: {} | SQL: {}", e.getMessage(), sql, e);
            return 0;
        } catch (Exception e) {
            log.error("未知错误: {}", e.getMessage(), e);
            return 0;
        }
    }


    // 将单次插入优化为批量定时插入，提升并发性
    // 多设备多行数据写入
    @Override
    //@Async("threadPoolTaskExecutor")
    public void insertDevicePropertyData(List<DeviceDataMessage> deviceDatas) {

        if (deviceDatas == null || deviceDatas.isEmpty()) {
            return;
        }

        // 可以定义为全局变量，通过clear方法清空，重复使用，就不用每秒都创建变量了
        List<String> deviceIds = new ArrayList<>();
        List<List<String>> measurementsList = new ArrayList<>();
        List<List<Object>> valuesList = new ArrayList<>();
        List<Long> timestamps = new ArrayList<>();
        List<List<TSDataType>> typesList = new ArrayList<>();

        for (DeviceDataMessage deviceData : deviceDatas){
            Map<String, Object> params = deviceData.getParams();
            if (params == null || params.isEmpty()) {
                continue;
            }
            List<String> measurements = new ArrayList<>(params.size());
            List<Object> values = new ArrayList<>(params.size());
            List<TSDataType> types = new ArrayList<>(params.size());
            deviceData.getParams().forEach((key, value) -> {
                measurements.add(key);
                values.add(value);
                types.add(determineDataType(value));
            });
            deviceIds.add(IotdbSqlUtils.buildTableName(deviceData.getProductKey(), deviceData.getChannelKey(), deviceData.getDeviceKey()));
            timestamps.add(deviceData.getTime());
            measurementsList.add(measurements);
            typesList.add(types);
            valuesList.add(values);
        }

        try {
            sessionPool.insertRecords(deviceIds, timestamps, measurementsList, typesList, valuesList);
        } catch (IoTDBConnectionException | StatementExecutionException e) {
            log.error("iotdb 插入数据失败",e);
        }
    }

    // PropertyType.java
    private TSDataType determineDataType(Object value) {
        if (value instanceof Integer) {
            return TSDataType.INT32;
        } else if (value instanceof Long) {
            return TSDataType.INT64;
        } else if (value instanceof Float) {
            return TSDataType.FLOAT;
        } else if (value instanceof Double) {
            return TSDataType.DOUBLE;
        } else if (value instanceof Boolean) {
            return TSDataType.BOOLEAN;
        } else if (value instanceof String) {
            return TSDataType.TEXT;
        }
        return TSDataType.DOUBLE;
    }

    private String buildQueryAllPropertiesSql(String tableName, DevicePropertyDataRequest request) {
        if (request.getAsc() == 0) {
            // 倒序查询
            return String.format(
                "SELECT * FROM %s WHERE time >= %d AND time <= %d ORDER BY time DESC LIMIT %d",
                tableName,
                request.getStartTime(),
                request.getEndTime(),
                request.getPageSize()
            );
        } else {
            // 正序查询
            if (request.getNextTime() != null && request.getNextTime() > 0) {
                // 不是第一页，使用nextTime作为新的startTime
                return String.format(
                    "SELECT * FROM %s WHERE time >= %d AND time <= %d ORDER BY time ASC LIMIT %d",
                    tableName,
                    request.getNextTime(),
                    request.getEndTime(),
                    request.getPageSize()
                );
            } else {
                // 第一页
                return String.format(
                    "SELECT * FROM %s WHERE time >= %d AND time <= %d ORDER BY time ASC LIMIT %d",
                    tableName,
                    request.getStartTime(),
                    request.getEndTime(),
                    request.getPageSize()
                );
            }
        }
    }


    private String buildQuerySpecificPropertiesSql(String tableName, String properties, DevicePropertyDataRequest request) {
        if (request.getAsc() == 0) {
            // 倒序查询
            return String.format(
                "SELECT %s FROM %s WHERE time >= %d AND time <= %d ORDER BY time DESC LIMIT %d",
                properties,
                tableName,
                request.getStartTime(),
                request.getEndTime(),
                request.getPageSize()
            );
        } else {
            // 正序查询
            if (request.getNextTime() != null && request.getNextTime() > 0) {
                // 不是第一页，使用nextTime作为新的startTime
                return String.format(
                    "SELECT %s FROM %s WHERE time >= %d AND time <= %d ORDER BY time ASC LIMIT %d",
                    properties,
                    tableName,
                    request.getNextTime(),
                    request.getEndTime(),
                    request.getPageSize()
                );
            } else {
                // 第一页
                return String.format(
                    "SELECT %s FROM %s WHERE time >= %d AND time <= %d ORDER BY time ASC LIMIT %d",
                    properties,
                    tableName,
                    request.getStartTime(),
                    request.getEndTime(),
                    request.getPageSize()
                );
            }
        }
    }
    
}
